package syan.kafkass2.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import syan.kafkass2.listener.KafkaConsumerListener;
import syan.kafkass2.listener.KafkaListenerJob;

import java.util.Collections;
import java.util.Properties;

/**
 * @author CHENTAIWEN
 * @create 2022/8/25 13:11
 */
@Slf4j
@Configuration
public class KafkaConfig{

    @Value("${spring.kafka.bootstrap-servers}")
    public String kafkaServer;

    @Value("${spring.kafka.consumer.key-deserializer}")
    public String kafkaConsumerKey;

    @Value("${spring.kafka.consumer.value-deserializer}")
    public String kafkaConsumerValue;

    @Value("${spring.kafka.consumer.enable-auto-commit}")
    public String kafkaAutoConfig;

    @Value("${spring.kafka.consumer.auto-commit-interval}")
    public String kafkaCommmitInterval;

    @Value("${spring.kafka.consumer.group-id}")
    public String kafkaGroupId;

    @Value("${spring.kafka.consumer.topic-name}")
    public String kafkaTopicName;
    @Value("${spring.kafka.consumer.auto-offset-reset}")
    public String kafkaResetConfig;

    @Autowired
    private KafkaConsumerListener kafkaConsumerListener;

    public static KafkaConsumer<String, String> kafkaConsumer;

    @Bean
    public void loadKafkaConfig() {
        Properties p = new Properties();
        p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, kafkaConsumerKey);
        p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, kafkaConsumerValue);
        p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaAutoConfig);
        p.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaCommmitInterval);
        p.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaGroupId);
        p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,kafkaResetConfig);
        p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,1);

        kafkaConsumer = new KafkaConsumer<String, String>(p);
        kafkaConsumer.subscribe(Collections.singletonList(kafkaTopicName));// 订阅消息
        log.error("消息订阅成功！kafka配置：" + p.toString());
        //启动消息监听线程
        KafkaListenerJob kafkaListenerJob = new KafkaListenerJob(kafkaConsumerListener);
        Thread t = new Thread(kafkaListenerJob);
        t.start();
    }
}
